
package org.apache.hadoop.examples.pagerank.leon;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;

import org.apache.hadoop.examples.pagerank.RankDistanceMeasure;
import org.apache.hadoop.examples.pagerank.RankReduceOutputCacheSwitch;
import org.apache.hadoop.examples.textpair.FirstPartitioner;
import org.apache.hadoop.examples.textpair.GroupComparator;
import org.apache.hadoop.examples.textpair.KeyComparator;
import org.apache.hadoop.examples.textpair.TextPair;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
 * 
 * @author leon
 * 
 */
public class PageRank {
	//aggreate the static data (key , value1|value2|value3|....)
	public static class StaticAggreationMapper extends MapReduceBase implements 
	         Mapper<LongWritable ,Text,Text,Text>{
        private List<String> tokenList=new ArrayList<String>();
        private Text outputKey=new Text();
        private Text outputValue=new Text();
        
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			tokenList.clear();
			String line=value.toString();
			StringTokenizer tokenizer=new StringTokenizer(line);
			while (tokenizer.hasMoreTokens())
				tokenList.add(tokenizer.nextToken());
			if(tokenList.size()>=2){
				String keyStr=tokenList.get(0);
				outputKey.set(keyStr.getBytes());
				String valueStr=tokenList.get(1);
				outputValue.set(valueStr.getBytes());
				output.collect(outputKey,outputValue);
			}
         			
		}
		
	}
	
	public static class StaticAggreationReducer extends MapReduceBase implements
	   Reducer<Text, Text, Text, Text> {
        String separator="|";
        Text value=new Text();
		public void reduce(Text key, Iterator<Text> values,
		OutputCollector<Text, Text> output, Reporter reporter)
		throws IOException {
			String valueStr=null;
			while (values.hasNext()) {
              valueStr=valueStr+separator+values.next();
			}
	      value.set(valueStr);     
	     output.collect(key, value);
     }
}
	
	 public static class ComputePrMapper extends MapReduceBase implements
	   Mapper<Text,Text,Text,FloatWritable>{

		/* Overriding methods
		 */
		 FloatWritable fw=new FloatWritable();
		 Text dest=new Text();
		@Override
		public void map(Text key, Text value,
				OutputCollector<Text, FloatWritable> output, Reporter reporter)
				throws IOException {
			String token=key.toString();
			String seperator="|";
		    float values =Float.parseFloat(value.toString());
			StringTokenizer stringtokenizer=new StringTokenizer(token,seperator);
			int count=stringtokenizer.countTokens();//count means the number of outside
			float pr=values/count;
			fw.set(pr);
			
			for(int i=0;i<count;i++){
			dest.set(stringtokenizer.nextToken());
	        output.collect(dest, fw);		
			}			
		}
		 
		 
		 
	 }
	 
	 public static class ComputePrReducer extends MapReduceBase implements 
	     Reducer<Text,FloatWritable,Text,FloatWritable>{
        float prtotal;
        float prtotals;
        FloatWritable value=new FloatWritable();
		@Override
		public void reduce(Text key, Iterator<FloatWritable> values,
				OutputCollector<Text, FloatWritable> output, Reporter reporter)
				throws IOException {
			
			while(values.hasNext()){				 
				 float pr=Float.parseFloat(values.next().toString());
				 prtotal+=pr;			
		       }
			prtotals=(float) (0.15+0.85*prtotal);
			value.set(prtotals);
			output.collect(key, value);
	 }
	 
	 
	 
//Test


	public static void main(String[] args) throws Exception {

		int iteration = -1;

		String inputPath = args[0];
		String outputPath = args[1];
		int specIteration = 0;
		if (args.length > 2) {
			specIteration = Integer.parseInt(args[2]);
		}
		
		int numNodes = 5;
		if (args.length > 3) {
			numNodes = Integer.parseInt(args[3]);
		}

		int reduceNum = 2;
		if (args.length > 4) {
			reduceNum = Integer.parseInt(args[4]);
		}
	    int mapNum=2;
	    if(args.length>5){
	    	mapNum=Integer.parseInt(args[5]);
	    }
	    
	    boolean fixpoint = false;
		if (args.length > 6) {
			fixpoint = Boolean.parseBoolean(args[6]);
		}

		System.out.println("specified iteration: " + specIteration);
		long start = System.currentTimeMillis();
   //pre-operation
		JobConf conf=new JobConf(PageRank.class);
		conf.setJobName("Pre-Operation");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		conf.setMapperClass(StaticAggreationMapper.class);
		conf.setReducerClass(StaticAggreationReducer.class);		
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));
		conf.setNumMapTasks(mapNum);
		conf.setNumReduceTasks(reduceNum);
		JobClient.runJob(conf);
	
		conf=new JobConf(PageRank.class);
	    conf.setJobName("PageRankCompute");
	    conf.setOutputKeyClass(Text.class);
    	conf.setOutputValueClass(FloatWritable.class);
		conf.setMapperClass(ComputePrMapper.class);
		conf.setReducerClass(ComputePrReducer.class);		
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));
		conf.setNumMapTasks(mapNum);
		conf.setNumReduceTasks(reduceNum);	
		if (fixpoint) {
			// set fix point logics
			conf.setFixpointCheck(true);
			conf.setDistanceMeasure(RankDistanceMeasure.class);
			conf.setLoopReduceOutputCacheSwitch(RankReduceOutputCacheSwitch.class);
			conf.setDistanceThreshold(1.0f);
		}	
		JobClient.runJob(conf);
		
		long end = System.currentTimeMillis();
		System.out.println("running time " + (end - start) / 1000 + "s");
	}
	 }
}
